package outputrocketmq

import (
	"context"
	"errors"
	"fmt"
	"log"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"github.com/bytedance/sonic"
	"github.com/tsaikd/gogstash/config"
	"github.com/tsaikd/gogstash/config/logevent"
)

const ModuleName = "rocketmq"

type OutputConfig struct {
	config.OutputConfig
	Rocketmq   string `json:"rocketmq" yaml:"rocketmq"`     // rocketmq to connect to
	Nameserver string `json:"nameserver" yaml:"nameserver"` //rocketmq nameservers
	Topic      string `json:"topic" yaml:"topic"`           // topic to publish to
	Groupname  string `json:"groupname" yaml:"groupname"`   //rocketmq groupname
	Retry      int    `json:"retry" yaml:"retry"`           //retry times

	client rocketmq.Producer
}

func DefaultOutputConfig() OutputConfig {
	return OutputConfig{
		OutputConfig: config.OutputConfig{
			CommonConfig: config.CommonConfig{
				Type: ModuleName,
			},
		},
		Retry: 3,
	}
}

func InitHandler(
	ctx context.Context,
	raw config.ConfigRaw,
	control config.Control,
) (config.TypeOutputConfig, error) {
	conf := DefaultOutputConfig()
	err := config.ReflectConfig(raw, &conf)
	if err != nil {
		return nil, err
	}
	if conf.Nameserver == "" {
		return nil, errors.New("Missing RocketMQ nameserver")
	}
	if conf.Topic == "" {
		return nil, errors.New("Missing RocketMQ topic")
	}
	if conf.Groupname == "" {
		return nil, errors.New("Missing RocketMQ groupname")
	}
	conf.Codec, err = config.GetCodecOrDefault(ctx, raw["codec"])
	if err != nil {
		return nil, err
	}
	conf.client, err = rocketmq.NewProducer(
		producer.WithNameServer([]string{conf.Nameserver}), // NameServer 地址
		producer.WithRetry(conf.Retry),                     // 重试次数
		producer.WithGroupName(conf.Groupname),             // 生产者组名
	)
	if err != nil {
		fmt.Printf("create producer error: %s\n", err.Error())
		return nil, err
	}

	// 启动 Producer
	err = conf.client.Start()
	if err != nil {
		fmt.Printf("start producer error: %s\n", err.Error())
		return nil, err
	}
	// defer conf.client.Shutdown()

	// sendMessageTest(conf.client)

	return &conf, nil
}

// func sendMessageTest(p rocketmq.Producer) {
// 	msg := &primitive.Message{
// 		Topic: "testTopic",
// 		Body:  []byte("Hello RocketMQ Go Client!"),
// 	}
// 	msg.WithTag("testTag")
// 	msg.WithKeys([]string{"testKey"})
// 	res, err := p.SendSync(context.Background(), msg)
// 	if err != nil {
// 		fmt.Printf("send message error: %s\n", err)
// 	} else {
// 		fmt.Printf("send message success: result=%s\n", res.String())
// 	}
// }

// Output event
func (t *OutputConfig) Output(ctx context.Context, event logevent.LogEvent) (err error) {
	raw, err := event.MarshalJSON()
	if err != nil {
		return err
	}
	um := make(map[string]interface{})
	err = sonic.Unmarshal(raw, &um)
	if err != nil {
		log.Println(err)
	}

	msg := &primitive.Message{
		Topic: t.Topic, // 主题
		Body:  []byte(um["message"].(string)),
	}

	res, err := t.client.SendSync(context.Background(), msg)
	if err != nil {
		fmt.Printf("send message error: %s\n", err)
	} else {
		fmt.Printf("send message success: result=%s\n", res.String())
	}
	return nil
}
